/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package edu.isi.pegasus.planner.subworkflow.partitioner;

import edu.isi.pegasus.planner.catalog.site.classes.SiteCatalogEntry;
import edu.isi.pegasus.planner.catalog.site.classes.SiteStore;
import edu.isi.pegasus.planner.classes.ADag;
import edu.isi.pegasus.planner.classes.PCRelation;
import edu.isi.pegasus.planner.classes.PegasusBag;
import edu.isi.pegasus.planner.classes.PegasusFile;
import edu.isi.pegasus.planner.namespace.Pegasus;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Vector;

/**
 *
 * This partitioner doesn't require estimator or scheduler.
 * pure partitioner itself does the work scheduler does. 
 * @author wchen
 */
public abstract class PurePartitioner implements PartitionerImplementation{
    protected ADag mDag ;
    protected PegasusBag mBag;
    protected LinkedList mQueue;
    protected Map mIDorLogicalID2Job;
    protected Map mSiteSize;
    protected LinkedList mSiteIndex;
    protected Map mID2LogicalID;
    protected Map mLogicalID2ID;
    protected Map mJob2Time;
    protected Map depthMap;
    protected Map mIndexJobList;
    protected Map mIndexFileList;
    protected Map mChildrenMap;
    protected Map mParentsMap;
    protected Map mJob2Site;
    public PurePartitioner(ADag mDag, PegasusBag mBag)
    {
        this.mDag = mDag;
        this.mBag = mBag;
        mQueue = new LinkedList();
        mIDorLogicalID2Job = new HashMap<String, edu.isi.pegasus.planner.classes.Job>();
        mSiteSize = new HashMap<Integer, Double>();
        mSiteIndex = new LinkedList<SiteCatalogEntry>();
        mID2LogicalID = new HashMap<String, String>();
        mLogicalID2ID = new HashMap<String, String>();
        mJob2Time = new HashMap<String, Double>();
        depthMap = new HashMap<String, Integer>();
        mIndexJobList = new HashMap<Integer, LinkedList>();
        mIndexFileList = new HashMap<Integer, LinkedList>();
        mChildrenMap = new HashMap<String, ArrayList>();
        mParentsMap   = new HashMap<String, ArrayList>();
        mJob2Site = new HashMap<String, SiteCatalogEntry>();
    }

    protected void getRel()
    {
        for(Iterator it = mDag.dagInfo.relations.iterator(); it.hasNext();)
        {
            PCRelation rel = (PCRelation) it.next();
            String parent = rel.parent;
            String child = rel.child;
            if(!mChildrenMap.containsKey(parent))
                mChildrenMap.put(parent, new ArrayList<String>());
            if(!mParentsMap.containsKey(child))
                mParentsMap.put(child, new ArrayList<String>());
            ArrayList clist = (ArrayList)mChildrenMap.get(parent);
            if(!clist.contains(child))
                clist.add(child);
            ArrayList plist = (ArrayList)mParentsMap.get(child);
            if(!plist.contains(parent))
                plist.add(parent);


        }
    }
    public void checkDaxWriteEnd()
    {

    }

    protected void translate()
    {
        for(Iterator it = mDag.jobIterator(); it.hasNext(); )
        {
            edu.isi.pegasus.planner.classes.Job job =
                    (edu.isi.pegasus.planner.classes.Job)it.next();

            mID2LogicalID.put(job.jobID, job.logicalId);
            mLogicalID2ID.put(job.logicalId, job.jobID);
            mIDorLogicalID2Job.put(job.logicalId, job);
            mIDorLogicalID2Job.put(job.jobID, job);
             String str = (String)job.vdsNS.get(Pegasus.RUNTIME_KEY);
            if(str==null)str = "0.0";
            double runtime = Double.parseDouble(str);
            mJob2Time.put(job.jobID, runtime);

        }
    }

    protected void addJob2Site(edu.isi.pegasus.planner.classes.Job dagJob, int siteIndex)
    {
        SiteCatalogEntry siteEntry = (SiteCatalogEntry)mSiteIndex.get(siteIndex);
        mJob2Site.put(dagJob, siteEntry);
       // double size_s= Double.parseDouble(siteEntry.getEnvironmentVariable("SITESIZE")) * 1e8;
       // int slot    = Integer.parseInt(siteEntry.getEnvironmentVariable("SLOT"));
        String siteName  = siteEntry.getSiteHandle();
        dagJob.setSiteHandle(siteName);
        if(!mIndexJobList.containsKey(siteIndex))
            mIndexJobList.put(siteIndex, new LinkedList());
        LinkedList jobList = (LinkedList)mIndexJobList.get(siteIndex);
        if(!jobList.contains(dagJob))
            jobList.add(dagJob);

        Map map = new HashMap();
        for(Iterator inFilesIt = dagJob.getInputFiles().iterator(); inFilesIt.hasNext();)
        {
            PegasusFile input = (PegasusFile)inFilesIt.next();
            String fileName = input.getLFN();
            if(input.getType()==1)
                continue;
            double size = input.getSize();
            if(!map.containsKey(fileName))
            {
                map.put(fileName, size);
            }
        }
        for(Iterator outFilesIt = dagJob.getOutputFiles().iterator(); outFilesIt.hasNext();)
        {
            PegasusFile output = (PegasusFile)outFilesIt.next();
            String fileName = output.getLFN();
            if (output.getType()==1)
                continue;
            double size = output.getSize();
            if(!map.containsKey(fileName))
            {
                map.put(fileName, size);
            }
        }


        Map fileList = (HashMap)mIndexFileList.get(siteIndex);
        if(fileList == null)
            mIndexFileList.put(siteIndex, new HashMap<String,Double>());
        for(Iterator it = map.keySet().iterator(); it.hasNext();)
        {
            String fileName = (String)it.next();
            double fileSize     = (Double)map.get(fileName);
            if(!fileList.containsKey(fileName))
            {
                fileList.put(fileName, fileSize);
            }
        }

        
    }
    protected edu.isi.pegasus.planner.classes.Job getDagJob(String jobname)
    {
        return (edu.isi.pegasus.planner.classes.Job)mIDorLogicalID2Job.get(jobname);
    }
    protected edu.isi.pegasus.planner.classes.Job getRear()
    {
      edu.isi.pegasus.planner.classes.Job rear =
              (edu.isi.pegasus.planner.classes.Job)mDag.vJobSubInfos.get(0);

      mQueue.clear();
      mQueue.addLast(rear.jobID);
      while(!mQueue.isEmpty())
      {
          String jobID = (String)mQueue.getLast();
          Vector vec = mDag.getChildren(jobID);
          if(vec.size() == 0)
          {
                rear = getDagJob(jobID);
                break;
          }
          else
          {
              mQueue.remove(jobID);
              mQueue.add(vec.get(0));
          }
      }
      return rear;
    }

    public void run(){
        getProperties();
        translate();
        getRel();
    }

    public PartitionResult getResult()
    {
        return null;
    }

    protected double getSizeMap(edu.isi.pegasus.planner.classes.Job dagJob, int siteIndex)
    {
       
        double sum = 0.0;
        Map map = new HashMap();
        for(Iterator inFilesIt = dagJob.getInputFiles().iterator(); inFilesIt.hasNext();)
        {
            PegasusFile input = (PegasusFile)inFilesIt.next();
            String fileName = input.getLFN();
            if(input.getType()==1)
                continue;
            double size = input.getSize();
            if(!map.containsKey(fileName))
            {
                map.put(fileName, size);
            }
        }
        for(Iterator outFilesIt = dagJob.getOutputFiles().iterator(); outFilesIt.hasNext();)
        {
            PegasusFile output = (PegasusFile)outFilesIt.next();
            String fileName = output.getLFN();
            if (output.getType()==1)
                continue;
            double size = output.getSize();
            if(!map.containsKey(fileName))
            {
                map.put(fileName, size);
            }
        }
        

        if(!mIndexFileList.containsKey(siteIndex))
        {
            mIndexFileList.put(siteIndex, new HashMap<String,Double>());
        }

        Map fileList = (HashMap)mIndexFileList.get(siteIndex);

        for(Iterator it = map.keySet().iterator(); it.hasNext();)
        {
            String fileName = (String)it.next();
            double size     = (Double)map.get(fileName);
            boolean hasFile = false;


            if(fileList.containsKey(fileName))
            {
                hasFile = true;
            }

            if (!hasFile)
            {
                sum += size;
            }
        }
        return sum;
    }

    
    protected double getSizeMap(edu.isi.pegasus.planner.classes.Job dagJob)
    {

        double sum = 0.0;
        Map map = new HashMap();
        for(Iterator inFilesIt = dagJob.getInputFiles().iterator(); inFilesIt.hasNext();)
        {
            PegasusFile input = (PegasusFile)inFilesIt.next();
            String fileName = input.getLFN();
            if(input.getType()==1)
                continue;
            double size = input.getSize();
            if(!map.containsKey(fileName))
            {
                map.put(fileName, size);
                sum += size;
            }
        }
        for(Iterator outFilesIt = dagJob.getOutputFiles().iterator(); outFilesIt.hasNext();)
        {
            PegasusFile output = (PegasusFile)outFilesIt.next();
            String fileName = output.getLFN();
            if (output.getType()==1)
                continue;
            double size = output.getSize();
            if(!map.containsKey(fileName))
            {
                map.put(fileName, size);
                sum += size;
            }
        }
        return sum;
    }
    protected double getSizeMap(int siteIndex)
    {
       if (!mSiteSize.containsKey(siteIndex))
       {
           double sum = 0.0;
           mSiteSize.put(siteIndex, sum);
           return sum;
       }
       double size = (Double)mSiteSize.get(siteIndex);
       return size;
    }

   protected void setSizeMap(int siteIndex, double size)
   {
       if(true)
       {
           mSiteSize.put(siteIndex, size);
       }
   }
    public ADag getDag()
    {
        return mDag;
    }
    protected String jobID2jobLogicalID(String jobID)
    {
        return (String)mID2LogicalID.get(jobID);
    }

     protected void getProperties()

    {
        try
        {
            mSiteIndex.clear();
            SiteStore mSite = mBag.getHandleToSiteStore();
            String reorder = mBag.getPegasusProperties().getProperty("pegasus.subworkflow.reorder");
            if( (reorder==null) || reorder.equals("no"))
            {
                for (Iterator it = mSite.entryIterator();it.hasNext();)
                {
                    SiteCatalogEntry siteEntry = (SiteCatalogEntry)it.next();
                    String site = siteEntry.getSiteHandle();
                    if(site.equals("local"))continue;
                    mSiteIndex.add(siteEntry);
                    System.out.println(site+" has been selected in PurePartitioner.java");
                }
            }
            else
            {

                for (Iterator it = mSite.entryIterator();it.hasNext();)
                {
                    SiteCatalogEntry siteEntry = (SiteCatalogEntry)it.next();
                    String site = siteEntry.getSiteHandle();
                    if(site.equals("local"))continue;
                    double limit = Double.parseDouble(siteEntry.getEnvironmentVariable("SITESIZE")) *1e8;
                    if(mSiteIndex.size() == 0)
                    {
                        mSiteIndex.add(siteEntry);
                    }
                    else
                    {
                        for(int i = 0; i < mSiteIndex.size(); i++)
                        {
                            SiteCatalogEntry curSiteEntry = (SiteCatalogEntry)mSiteIndex.get(i);
                            double curLimit = Double.parseDouble(curSiteEntry.getEnvironmentVariable("SITESIZE")) *1e8;
                            if(curLimit < limit)
                            {
                                mSiteIndex.add(i, siteEntry);
                                break;
                            }
                            if( i == (mSiteIndex.size() -1) )
                            {
                                mSiteIndex.addLast(siteEntry);
                                break;
                            }

                        }
                    }

                }
                for(int i = 0 ; i < mSiteIndex.size(); i ++)
                {
                   SiteCatalogEntry curSiteEntry = (SiteCatalogEntry)mSiteIndex.get(i);
                   double curLimit = Double.parseDouble(curSiteEntry.getEnvironmentVariable("SITESIZE"));
                   System.out.println("site: " + curSiteEntry.getSiteHandle() +" " + curLimit);
                }

            }

        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }


}
